package streaming.api.sink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import utils.PropertiesReader;

import java.util.Properties;

/**
 * kafka -> kafka
 * 数据来源： kafka topic[csvTest1]
 * Sink目标： kafka topic[csvTest2]
 */
public class SinkTest1_kafka {

    private static String kafkaServers = PropertiesReader.get("default.kafka.servers");
    private static String topicFrom = PropertiesReader.get("default.kafka.topic.csv.A");
    private static String topicTo = PropertiesReader.get("default.kafka.topic.csv.B");

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Properties props = new Properties();
        props.setProperty("bootstrap.servers", kafkaServers);
        props.setProperty("group.id", "flink-consumer-SinkTest1_kafka");

        // 从Kafka中读取数据
        DataStream<String> inputStream = env.addSource(new FlinkKafkaConsumer(topicFrom, new SimpleStringSchema(), props));
        // 将空格转成逗号
        DataStream<String> dataStream = inputStream.map(line -> {
            return line.replace("", ",");
        });
        // 将数据写入Kafka
        dataStream.addSink( new FlinkKafkaProducer<String>(kafkaServers, topicTo, new SimpleStringSchema()));
        env.execute();
    }
}
